package org.jetlinks.pro.things.impl.connector;

import lombok.EqualsAndHashCode;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.pro.streaming.Computer;
import org.jetlinks.pro.streaming.Streaming;
import org.jetlinks.pro.things.ThingConstants;
import reactor.core.Disposable;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.function.Function3;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import java.util.function.Function;

/**
 * 物消息流式计算
 *
 * @param <T> 计算结果
 */
@EqualsAndHashCode(of = "identity")
public class ThingMessageComputeStreaming<T> implements Streaming<ThingMessage, T, T> {


    //消息传递处理器，通过同步流输入输出循环实现任务的发布-订阅。
    private final EmitterProcessor<T> processor = EmitterProcessor.create(false);

    //用于下游数据订阅的输出
    private final FluxSink<T> output = processor.sink(FluxSink.OverflowStrategy.BUFFER);

    //可以取消/释放任务或资源的容器。
    private Disposable streaming;

    //用于接收物消息
    private FluxSink<ThingMessage> input;

    //物消息流式计算的唯一标识
    private Object identity;

    /**
     *
     * @param identity 物消息流式计算的唯一标识
     * @param computer 计算器
     * @param converter 消息转换器函数
     * @param resultConverter 计算结果转换器函数
     * @param <V> 结果计算流
     * @param <T> 计算结果
     * @param <R> 属性计算结果对象
     * @return 物消息流式计算对象
     */
    public static <V, T, R> ThingMessageComputeStreaming<R> create(Object identity,
                                                                   Computer<ThingMessage, V> computer,
                                                                   Function<Flux<V>, Flux<T>> converter,
                                                                   Function3<ThingId, Set<String>, Object, R> resultConverter) {
        return ThingMessageComputeStreaming
            .create(identity,
                    //计算消息结果后，做转换处理，例如窗口处理
                    flux -> flux.flatMap(computer::compute).as(converter),
                    resultConverter);
    }

    /**
     *
     * @param identity 物消息流式计算的唯一标识
     * @param computer 计算器
     * @param converter 消息转换器函数
     * @param <T> 计算结果
     * @param <R> 属性计算结果对象
     * @return 物消息流式计算对象
     */
    public static <T, R> ThingMessageComputeStreaming<R> create(Object identity,
                                                                Function<Flux<ThingMessage>, Flux<T>> computer,
                                                                Function3<ThingId, Set<String>, Object, R> converter) {
        ThingMessageComputeStreaming<R> streaming = new ThingMessageComputeStreaming<>();
        streaming.identity = identity;
        streaming.streaming = Flux
            .<ThingMessage>create(sink -> streaming.input = sink)
            //按物ID分组
            .groupBy(msg -> ThingId.of(msg.getThingType(), msg.getThingId()), Integer.MAX_VALUE)
            .flatMap(group -> {
                @SuppressWarnings("all")
                ThingId thingId = group.key();
                //记录连接器跟踪信息
                Set<String> traces = ConcurrentHashMap.newKeySet();

                return computer
                    .apply(group
                               .doOnNext(msg -> {
                                   //合并消息跟踪header，通常用于处理连接器是否存在递归
                                   msg.getHeader(ThingConstants.connectorTrace)
                                      .ifPresent(traces::addAll);
                               }))
                    .doOnNext(value -> {
                        R message = converter.apply(thingId, traces, value);
                        if (message != null) {
                            streaming.output.next(message);
                        }
                    })
                    .onErrorResume(err -> Mono.empty())
                    .then();
            }, Integer.MAX_VALUE)
            .subscribe();

        return streaming;
    }

    /**
     * 输入一个物消息到计算规则中
     *
     * @param data 数据
     * @return 计算结果
     */
    @Override
    public Mono<T> compute(ThingMessage data) {
        if (processor.hasDownstreams()) {
            input.next(data);
        }
        return Mono.empty();
    }

    /**
     * 获取计算输出结果流
     *
     * @return 计算结果
     */
    @Override
    public Flux<T> output() {
        return processor;
    }

    //释放资源
    @Override
    public void dispose() {
        processor.onComplete();
        output.complete();
        streaming.dispose();
        input.complete();
    }

}